package com.sh.data.engine.application.service.datadev;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.sh.data.engine.application.model.dto.datadev.FlinkResourceResponse;
import com.sh.data.engine.application.model.dto.datadev.UploadFlinkResourceResponse;
import com.sh.data.engine.application.model.request.datadev.SaveResourceRequest;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.base.model.FileUploadResultDomain;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.service.FileService;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeConfigResourceDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.util.ConvertUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

/**
 * @author: mengzheng.mhc
 * @date: 2024/8/12 9:59
 */
@Service
@Slf4j
public class FlinkResourceApplicationService {

    @Value("${data.engine.dataDevResource}")
    private String dataDevResource;

    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;

    @Autowired
    private FileService fileService;

    public UploadFlinkResourceResponse uploadResource(MultipartFile multipartFile) {
        // 如果文件不是以.jar结尾
        String fileName = multipartFile.getOriginalFilename();
        if (!fileName.toLowerCase().endsWith(".jar")) {
            throw new BusinessException("文件格式错误");
        }
        FileUploadResultDomain upload = null;
        String dataDevResource = this.dataDevResource;
        try {
            upload = fileService.upload(multipartFile, dataDevResource);
        } catch (Exception e) {
            throw new BusinessException("上传文件失败");
        }
        return ConvertUtil.copyProperties(upload, UploadFlinkResourceResponse.class);
    }

    public boolean saveFlinkResource(SaveResourceRequest saveResourceRequest) {
        if (null == saveResourceRequest.getPid()) {
            saveResourceRequest.setPid(2L);
        }
        FlinkNodeConfigResourceDomain saveResourceParam =
            Convert.convert(FlinkNodeConfigResourceDomain.class, saveResourceRequest);
        return flinkNodeConfigService.saveFlinkResource(saveResourceParam);
    }

    public FlinkResourceResponse getFlinkResourceResponseDetail(Long nodeId) {
        FlinkNodeConfigResourceDomain activeFlinkResource =
            flinkNodeConfigService.getActiveFlinkResource(nodeId);
        return Convert.convert(FlinkResourceResponse.class, activeFlinkResource);
    }

    public PageResult<FlinkResourceResponse> getFlinkResourceResponseList(Long nodeId, Integer pageNum, Integer pageSize) {
        IPage<FlinkResourceResponse> flinkResourcePage =
            flinkNodeConfigService
                .getFlinkResourcePage(nodeId, pageNum, pageSize)
                .convert(e -> Convert.convert(FlinkResourceResponse.class, e));

        return PageResult.copyPropertiesAndConvert(flinkResourcePage, FlinkResourceResponse.class);
    }

    public boolean editFlinkResourceByNodeId(Long nodeId, String resourceName) {
        flinkNodeConfigService.editFlinkResourceByNodeId(nodeId, resourceName);
        return true;
    }

    public boolean dropFlinkResourceById(Long nodeId) {
        flinkNodeConfigService.deleteResourceById(nodeId);
        return true;
    }
}
